package com.upcoming.data.storm.caller;

import com.upcoming.data.storm.util.MyUtil;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

/**
 * 作者：tianjiayuan
 * 创建时间：2018-05-17 13:26
 * 类描述：
 * 修改人：
 * 修改时间：
 */
public class CallLogCounterBolt implements IRichBolt {

    private Map<String, Integer> counterMap;

    private OutputCollector collector;

    public CallLogCounterBolt() {
        MyUtil.outLog2NC(this, "new CallLogCounterBolt()");
    }


    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        MyUtil.outLog2NC(this, "Bolt1.prepare()");
        this.counterMap = new HashMap<String, Integer>();
        this.collector = outputCollector;
    }

    public void execute(Tuple tuple) {
        String call = tuple.getString(0);
        Integer duration = tuple.getInteger(1);
        if (!counterMap.containsKey(call)) {
            counterMap.put(call, 1);
            MyUtil.outLog2NC(this, call + ":" + 1);
        } else {
            Integer c = counterMap.get(call);
            counterMap.put(call, c+1);
            MyUtil.outLog2NC(this, call + ":" + (c+1));
        }
        this.collector.ack(tuple);
    }

    public void cleanup() {

    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
